package kproduct

import (
	"context"
	"github.com/Shopify/sarama"
	"github.com/rcrowley/go-metrics"
	"log"
	"strings"
	"time"
)

type AsyncKafkaProduct struct {
	Brokers 	string
	
	producer 	sarama.AsyncProducer
	cancel 		context.CancelFunc
	logFunc 	func(v ...interface{})
}

func (k *AsyncKafkaProduct) Close()  {
	k.cancel()
	return
}

func NewAsyncKafkaProduct(brokers string, cancel context.CancelFunc, useNilMetrics bool, logFunc func(v ...interface{})) *AsyncKafkaProduct {
	metrics.UseNilMetrics = useNilMetrics
	return &AsyncKafkaProduct{
		Brokers: brokers,
		cancel: cancel,
		logFunc: logFunc,
	}
}

func (k *AsyncKafkaProduct) SetLogger(logger *log.Logger)  {
	sarama.Logger = logger
	return
}

func (k *AsyncKafkaProduct) Start(ctx context.Context) error {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Return.Successes = true
	config.Producer.Timeout = 5 * time.Second

	p, err := sarama.NewAsyncProducer(strings.Split(k.Brokers, ","), config)
	if err != nil {
		return err
	}
	k.producer = p
	
	k.errLog(ctx)
	return nil
}

func (k *AsyncKafkaProduct) errLog(ctx context.Context) {
	go func(p sarama.AsyncProducer) {
		errors := p.Errors()
		success := p.Successes()

		for {
			select {
			case err := <- errors:
				if err != nil {
					if k.logFunc != nil {
						k.logFunc("KafkaAsyncProduct: ", err.Error())
					}
				}
			case <- success:

			case <- ctx.Done():
				if k.logFunc != nil {
					k.logFunc("KafkaAsyncProduct: Done stop kafka producer, close producer resource")
				}
				p.Close()
				return
			}
		}
	}(k.producer)

	return
}

func (k *AsyncKafkaProduct) SendMessage(msg *sarama.ProducerMessage) (partition int32, offset int64, err error) {
	k.producer.Input() <- msg
	return 0, 0, nil
}

func (k *AsyncKafkaProduct) SendMessages(msgs []*sarama.ProducerMessage) error {

	for _, m := range msgs {
		k.producer.Input() <- m
	}
	return nil
}